package org.acme.cluster.service;

import com.baomidou.mybatisplus.extension.service.impl.ServiceImpl;
import org.acme.cluster.dao.SysSmsDao;
import org.acme.cluster.lock.DbLock;
import org.acme.cluster.po.SysSms;
import org.acme.cluster.utils.AddressUtils;
import org.jgroups.Address;
import org.jgroups.View;
import org.jgroups.util.UUID;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.BeansException;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.context.ApplicationContext;
import org.springframework.context.ApplicationContextAware;
import org.springframework.stereotype.Service;

import javax.annotation.Resource;
import javax.sql.DataSource;
import java.util.ArrayList;
import java.util.Date;
import java.util.List;

@Service
public class SysSmsServiceImpl extends ServiceImpl<SysSmsDao, SysSms> implements SysSmsService, ApplicationContextAware {

    // 重试间隔
    private static final int retryInterval = 60000;
    Logger logger = LoggerFactory.getLogger(getClass());
    //最大重试次数
    @Value("${sms.max-retry-count:10}")
    private int maxRetryCount;
    //批量处理的最大数量
    @Value("${sms.max-batch-size:50}")
    private int maxBatchSize;
    @Resource
    private SysClusterService sysClusterService;
    @Resource
    private SysSmsDispatchService sysSmsDispatchService;
    @Resource
    private DataSource dataSource;
    //spring容器
    private ApplicationContext applicationContext;

    /*****
     * 根据服务标识查询短信
     * @param serverKey
     * @param excludeIds
     * @return
     */
    @Override
    public List<SysSms> findByServerKey(String serverKey, List<String> excludeIds) {
        if (excludeIds.isEmpty()) {
            return getBaseMapper().getSmsListByServerKey(serverKey, maxRetryCount, maxBatchSize);
        }
        return getBaseMapper().getSmsListByServerKeyExceptRetryIdList(serverKey, excludeIds, maxRetryCount, maxBatchSize);
    }

    /***
     * 发送短信信息
     * @param sysSms
     * @return
     */
    @Override
    public boolean sendSms(SysSms sysSms) {
        try {
            int retryCount = sysSms.getRetryCount() + 1;
            sysSms.setRetryCount(retryCount);
            sysSms.setNextRetryTime(new Date(System.currentTimeMillis() + (long) retryInterval * retryCount));
            sysSms.setStatus(false);
            sysSms.setErrorMsg("发送失败");
            updateById(sysSms);
        } catch (Exception e) {
            logger.warn("短信发送失败: " + " id: " + sysSms.getId());
        }
        logger.warn("短信发送失败: " + " id: " + sysSms.getId());
        return false;
    }


    /****
     * 添加发送短信
     * @param sysSms
     * @return
     */
    @Override
    public boolean addSms(SysSms sysSms) {
        sysSms.setServerKey(sysClusterService.getUuid().toStringLong());
        sysSms.setRetryCount(0);
        sysSms.setCreateTime(new Date());
        sysSms.setUpdateTime(new Date());
        sysSms.setStatus(false);
        return save(sysSms) && sysSmsDispatchService.refreshTask();
    }

    /****
     * 批量添加发送短信，此处需要实现分布式协调。
     * @param sysSmsList
     * @return
     */
    @Override
    public boolean addSmsBatch(List<SysSms> sysSmsList) throws Exception {
        View view = this.sysSmsDispatchService.getView();
        if (sysSmsList.size() > maxBatchSize) {
            int index = sysSmsList.size() / view.getMembers().size();
            for (int i = 0; i < view.getMembers().size(); i++) {
                Address address = view.getMembers().get(i);
                List<SysSms> subList = new ArrayList<>();
                if (i == view.getMembers().size() - 1) {
                    subList = sysSmsList.subList(i * index, sysSmsList.size());
                } else {
                    subList = sysSmsList.subList(i * index, (i + 1) * index);
                }
                for (SysSms sysSms : subList) {
                    sysSms.setServerKey(((UUID) address).toStringLong());
                    sysSms.setRetryCount(0);
                    sysSms.setCreateTime(new Date());
                    sysSms.setUpdateTime(new Date());
                    sysSms.setStatus(false);
                    this.save(sysSms);
                }
            }
            return this.sysSmsDispatchService.refreshALLTask();
        } else {
            for (SysSms sysSms : sysSmsList) {
                sysSms.setServerKey(sysClusterService.getUuid().toStringLong());
                sysSms.setRetryCount(0);
                sysSms.setCreateTime(new Date());
                sysSms.setUpdateTime(new Date());
                sysSms.setStatus(false);
                save(sysSms);
            }
            return this.sysSmsDispatchService.refreshTask();
        }
    }

    /***
     * 获取最大重试次数
     * @return
     */
    @Override
    public int getMaxRetryCount() {
        return this.maxRetryCount;
    }

    /****
     * 节点掉线后，自动把节点任务做分片处理，把任务均匀分配到各个节点。
     * @param activeServers 目前活跃的节点列表
     * @param currentServer 当前节点
     * @param failedServers 掉线的节点列表
     */
    @Override
    public void updateSmsStatusToAll(List<Address> activeServers, Address currentServer, List<Address> failedServers) {
        DbLock lock = null;
        try {
            lock = new DbLock(dataSource);
            lock.lock();
            List<String> failedServerKeys = AddressUtils.getAddresses(failedServers);
            List<String> sysSmsList = getBaseMapper().getSmsListByServerKeys(failedServerKeys, getMaxRetryCount());
            if (sysSmsList != null && sysSmsList.size() > 0) {
                if (sysSmsList.size() > maxBatchSize) {
                    //分片处理，把任务均匀分配到各个节点。
                    int index = sysSmsList.size() / activeServers.size();
                    for (int i = 0; i < activeServers.size(); i++) {
                        if (i == activeServers.size() - 1) {
                            List<String> subList = sysSmsList.subList(i * index, sysSmsList.size());
                            getBaseMapper().updateSmsServerKeyByIdList(subList, ((UUID) activeServers.get(i)).toStringLong(), getMaxRetryCount());
                        } else {
                            List<String> subList = sysSmsList.subList(i * index, (i + 1) * index);
                            getBaseMapper().updateSmsServerKeyByIdList(subList, ((UUID) activeServers.get(i)).toStringLong(), getMaxRetryCount());
                        }
                    }
                } else {
                    //不做分片，把所有的任务都给当前节点。
                    getBaseMapper().updateSmsStatusByServerKeys(failedServerKeys, ((UUID) currentServer).toStringLong(), getMaxRetryCount());
                }
            }
        } catch (Exception e) {
            logger.warn("更新短信服务节点失败: " + e.getMessage());
        } finally {
            lock.unlock();
        }
    }

    /****
     * 集群节点刚启动的时候，把所有未完成的任务全部拉起
     * @param serverKey
     */
    @Override
    public void updateAllSmsServerKey(String serverKey) {
        DbLock lock = null;
        try {
            lock = new DbLock(dataSource);
            lock.lock();
            getBaseMapper().updateALLSmsServerKey(serverKey, getMaxRetryCount());
        } catch (Exception e) {
            logger.warn("更新短信服务节点失败: " + e.getMessage());
        } finally {
            lock.unlock();
        }
    }

    /***
     * spring容器服务类
     * @param applicationContext
     * @throws BeansException
     */
    @Override
    public void setApplicationContext(ApplicationContext applicationContext) throws BeansException {
        this.applicationContext = applicationContext;
    }
}
